package cn.home.bootstrap.rpc;

import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Envelope;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeoutException;

/**
 *
 */
@Slf4j
public class RPCClient {
  private final static String IP = "127.0.0.1";
  private final static int  PORT = 5672;
  private Connection connection;
  private Channel channel;
  private String requestQueueName = "rpc_queue";
  private String replyQueueName;

  public RPCClient() throws IOException, TimeoutException {
      ConnectionFactory factory = new ConnectionFactory();
      factory.setHost(IP);
      factory.setPort(PORT);
      connection = factory.newConnection();
      channel = connection.createChannel();
      //RPC调用结果队列
      replyQueueName = channel.queueDeclare().getQueue();
  }
  public static void main(String[] argv) {
    RPCClient fibonacciRpc = null;
    String response = null;
    try {
      fibonacciRpc = new RPCClient();

     log.info(" [x] Requesting fib(30)");
     response = fibonacciRpc.call("30");
     log.info(" [.] RPC result Got :{}", response);
    }
    catch  (IOException | TimeoutException | InterruptedException e) {
      e.printStackTrace();
    }
    finally {
      if (fibonacciRpc!= null) {
        try {
          fibonacciRpc.close();
        }
        catch (IOException _ignore) {}
      }
    }
  }
  /**
   * @param message
   * @return
   * @throws IOException
   * @throws InterruptedException
   */
  public String call(String message) throws IOException, InterruptedException {
    //请求id，保证幂等
    final String corrId = UUID.randomUUID().toString();
    AMQP.BasicProperties props = new AMQP.BasicProperties
            .Builder()
            .correlationId(corrId)
            .replyTo(replyQueueName)
            .build();
   //发起RPC调用
    channel.basicPublish("", requestQueueName, props, message.getBytes("UTF-8"));
    final BlockingQueue<String> response = new ArrayBlockingQueue<String>(1);
    channel.basicConsume(replyQueueName, true, new DefaultConsumer(channel) {
      @Override
      public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        if (properties.getCorrelationId().equals(corrId)) {
          response.offer(new String(body, "UTF-8"));
        }
      }
    });
    return response.take();
  }

  /**
   * @throws IOException
   */
  public void close() throws IOException {
    connection.close();
  }
}

